package blue.mqtt4j.server.net;

import blue.mqtt4j.server.topic.TopicChannel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttSubscribePayload;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Jin Zheng
 * @since 2020-03-22
 */
public class MqttSubscribeHandler extends SimpleChannelInboundHandler<MqttSubscribeMessage>
{
	private static Logger logger = LoggerFactory.getLogger(MqttSubscribeHandler.class);

	private final TopicChannel topicChannel;

	public MqttSubscribeHandler(TopicChannel topicChannel)
	{
		this.topicChannel = topicChannel;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttSubscribeMessage message) throws Exception
	{
		MqttFixedHeader reqHeader = message.fixedHeader();
		MqttMessageIdVariableHeader variableHeader = message.variableHeader();
		MqttSubscribePayload payload = message.payload();
		logger.info("header: {}", reqHeader);
		logger.info("variableHeader: {}", variableHeader);
		logger.info("payload: {}", payload);

		List<MqttQoS> qosList = new ArrayList<>();
		for (MqttTopicSubscription subscription : payload.topicSubscriptions())
		{
			qosList.add(subscription.qualityOfService());
			topicChannel.add(subscription.topicName(), ctx.channel());
		}

		MqttSubAckMessage ackMessage = MqttMessageBuilders.subAck()
				.addGrantedQoses(qosList.toArray(new MqttQoS[0]))
				.packetId((short)1)
				.build();
		ctx.writeAndFlush(ackMessage);
	}
}
